package com.k8.cluster.loadBalanceImpl;

import com.k8.rpc.cluster.loadBalance.LoadBalance;
import com.k8.rpc.directory.Directory;
import com.k8.rpc.invoke.Invoker;
import com.k8.rpc.util.HashUtils;

import java.util.*;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @Author: k8
 * @CreateTime: 2025-03-20
 * @Version: 1.0
 * 要想起作用ClusterInvoker就必须是FailFast
 * <p>
 * 用于在会话开启的时候始终使用同一个实例
 */
public abstract class AbstractConsistentHashLoadBalance<T> implements LoadBalance<T> {
    private Directory<T> directory;
    private boolean pushed;
    private final TreeMap<Integer, Invoker<T>> hashRing = new TreeMap<>();
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(false);
    private final int virtualNodeCount = 1000;
    //因为具体存储hash轮的是TreeMap，当获取大于等于它的第一个key不存在时，那么就应该拿最小的hash值
    private volatile int minHash;

    @Override
    public void pushInvokersDirectory(Directory<T> directory) {
        if (pushed) return;
        synchronized (hashRing) {
            if (pushed) return;
            pushed = true;
            this.directory = directory;
            minHash = Integer.MAX_VALUE;
            for (Invoker<T> invoker : directory.list()) {
                for (int i = 1; i <= virtualNodeCount; ++i) {
                    String hashS = String.valueOf(invoker.hashCode() + i);
                    int hash = HashUtils.murmurHash32(hashS);
                    int j = 0;
                    while (hashRing.containsKey(hash)) {
                        hash = HashUtils.murmurHash32(hashS + j);
                        j++;
                    }
                    if (hash < minHash) minHash = hash;
                    hashRing.put(hash, invoker);
                }
            }
            directory.addInvokerAddListener((invokers) -> {
                try {
                    lock.writeLock().lock();
                    for (Invoker<?> invoker : invokers) {
                        Invoker<T> invoker1 = (Invoker<T>) invoker;
                        for (int i = 1; i <= virtualNodeCount; ++i) {
                            String hashS = String.valueOf(invoker1.hashCode() + i);
                            int hash = HashUtils.murmurHash32(hashS);
                            int j = 0;
                            while (hashRing.containsKey(hash)) {
                                hash = HashUtils.murmurHash32(hashS + j);
                                j++;
                            }
                            if (hash < minHash) minHash = hash;
                            hashRing.put(hash, invoker1);
                        }
                    }
                } finally {
                    lock.writeLock().unlock();
                }
            });

        }
    }

    public Invoker<T> getInvoker(String key) {
        int hash32 = HashUtils.murmurHash32(key);
        try {
            lock.readLock().lock();
            SortedMap<Integer, Invoker<T>> integerInvokerSortedMap = hashRing.tailMap(hash32);
            Integer hash = minHash;
            if (!integerInvokerSortedMap.isEmpty()) hash = integerInvokerSortedMap.firstKey();
            return hashRing.get(hash);
        } finally {
            lock.readLock().unlock();
        }
    }

    @Override
    public Invoker<T> nextInvoker() {
        throw new IllegalArgumentException("AbstractConsistentHashLoadBalance need hash key.");
    }

    @Override
    public int getSize() {
        return directory.size();
    }


}
